package com.qinggeng.framework.config.mq;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.openservices.ons.api.*;
import com.qinggeng.common.exception.CustomException;
import com.qinggeng.framework.config.mq.core.GsdMessageService;
import com.qinggeng.framework.config.mq.model.MessageModel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.Date;

@Component
@Slf4j
public class MQSendUtils {

    @Autowired
    private Producer producer;

    @Autowired
    private GsdMessageService gsdMessageService;

    @Value("${rocketmq.topic}")
    private String topic;

    @Value("${rocketmq.delayedTopic}")
    private String delayedTopic;


    //同步发送普通消息
    public void sendMessage(MessageModel messageModel) {
        checkMessgae(messageModel);
        Message message = getSendMessage(messageModel);
        SendResult sendResult = null;
        try {
            sendResult = producer.send(message);
            if (null != sendResult) {
                success(message);
            }
        } catch (Exception e) {
            error(message, e);
        }
    }

    //获取普通模型
    private Message getSendMessage(MessageModel messageModel) {
        Message message = new Message();
        String msgId = messageModel.getMessageId().toString();
        message.setMsgID(msgId);
        String key = messageModel.getType() + ":" + msgId;
        message.setKey(key);
        message.setTopic(topic);
        message.setBody(JSONObject.toJSONBytes(messageModel));
        boolean bl = messageWarehousing(messageModel);
        if (bl == false) {
            throw new CustomException("消息入库失败");
        }
        return message;
    }

    //消息入库
    private boolean messageWarehousing(MessageModel messageModel) {
        return gsdMessageService.insertGsdMessage(messageModel);
    }

    //异步发送
    public void sendAsynMessage(MessageModel messageModel) {
        checkMessgae(messageModel);
        Message message = getSendMessage(messageModel);
        SendResult sendResult = null;
        producer.sendAsync(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {

                log.info("发送成功: {} ", sendResult);
            }

            @Override
            public void onException(OnExceptionContext onExceptionContext) {
                log.info("发送失败: {} ", sendResult);

            }
        });
    }

    //延时发送
    public void sendDelayMessage(MessageModel messageModel, Long seconds) {
        checkMessgae(messageModel);
        /*if (seconds <= 5) {
            throw new CustomException("延时小于5秒,请发送普通消息");
        }*/
        Message message = getSendMessage(messageModel);
        message.setTopic(delayedTopic);
        long millis = seconds * 1000;
        long delayTime = System.currentTimeMillis() + millis;
        SimpleDateFormat sdf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String sd = sdf.format(new Date(delayTime));
        log.info("接收延时消息时间 " + sd);

        message.setStartDeliverTime(delayTime);
        SendResult sendResult = null;
        try {
            sendResult = producer.send(message);
            if (null != sendResult) {
                success(message);
                log.info("发送延时消息时间 " + DateUtil.formatDateTime(new Date()));
            }
        } catch (Exception e) {
            error(message, e);
        }
    }

    //定时发送      LocalDateTime localDateTime = LocalDateTime.of(2020,11,17,15,12);
    public void sendTimingMessage(MessageModel messageModel, LocalDateTime localDateTime) {

        checkMessgae(messageModel);

        Message message = getSendMessage(messageModel);
        message.setTopic(delayedTopic);
        long timeStamp = localDateTime.toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
        message.setStartDeliverTime(timeStamp);
        SendResult sendResult = null;
        try {
            sendResult = producer.send(message);
            if (null != sendResult) {
                success(message);
               log.info("发送消息时间 " + DateUtil.formatDateTime(new Date()));
            }
        } catch (Exception e) {
            error(message, e);
        }
    }

    private void checkMessgae(MessageModel messageModel) {
        log.info("checkMessage"+messageModel);
        Long messageId = messageModel.getMessageId();
        if (null == messageId) {
            messageModel.setMessageId(IdUtil.createSnowflake(1L, 1L).nextId());
        }
        Integer type = messageModel.getType();
        if (null == type) {
            throw new CustomException("消息type不能爲空");
        }
    }

    /**
     * 成功日志打印
     *
     * @param msg
     * @param
     */
    private void success(Message msg) {
        log.info("发送MQ消息成功 -- Topic:{} , Key:{}, tag:{}, body:{}"
                , msg.getTopic(), msg.getKey(), msg.getTag(), new String(msg.getBody()));
    }

    /**
     * 异常日志打印
     *
     * @param msg
     * @param e
     */
    private void error(Message msg, Exception e) {
        log.error("发送MQ消息失败-- Topic:{}, Key:{}, tag:{}, body:{}"
                , msg.getTopic(), msg.getKey(), msg.getTag(), new String(msg.getBody()));
        log.error("errorMsg", e);
    }

}
